package com.du.integration.server.config;

import cn.hutool.crypto.digest.MD5;
import com.du.integration.server.handle.MqttMsgHandle;
import com.du.integration.server.handle.MqttMsgHandleFactory;
import com.du.integration.server.model.entity.MqttMsg;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Optional;

/**
 * @Desc :
 * @Author: shuangdu
 * @date : 2024/1/5
 */
@Slf4j
public class CustomMqttClient {

    private MqttOption option;

    private MqttClient mqttClient;

    public CustomMqttClient(MqttOption option) {
        this.option = option;
    }

    public MqttConnectionOptions makeOption(MqttOption option) {
        MqttConnectionOptions ops = new MqttConnectionOptions();
        ops.setUserName(option.getUsername());
        ops.setPassword(MD5.create().digestHex(option.getPassword()).getBytes());
        ops.setCleanStart(Optional.ofNullable(option.getClearSession()).orElse(Boolean.FALSE));
        ops.setKeepAliveInterval(Optional.ofNullable(option.getKeepAlive()).orElse(30));
        ops.setConnectionTimeout(Optional.ofNullable(option.getTimeout()).orElse(10));
        return ops;
    }

    public void connect() throws Exception{
        if (mqttClient == null) {
            mqttClient = new MqttClient(option.getHost(), option.getClientId(), new MemoryPersistence());
            // 回调
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void disconnected(MqttDisconnectResponse disconnectResponse) {
                    log.info(">>>断开链接 {}", disconnectResponse);
                }

                @Override
                public void mqttErrorOccurred(MqttException exception) {
                    log.info(">>>出现异常 ", exception);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    log.info("接收到消息, topic: {}, msg: {}", topic, message);
                    MqttMsgHandle handle = MqttMsgHandleFactory.getHandle(topic);
                    if (handle != null) {
                        log.info("处理消息：{}", handle.getClass().getSimpleName());
                        handle.handle(topic, message);
                    }
                }

                @Override
                public void deliveryComplete(IMqttToken token) {
                    log.info(">>>deliveryComplete {}", token);
                }

                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    log.info(">>>链接成功 {}, 是否重连{}", serverURI, reconnect);
                }

                @Override
                public void authPacketArrived(int reasonCode, MqttProperties properties) {
                    log.info(">>>authPacketArrived {}- {}", reasonCode, properties);
                }
            });
        }

        mqttClient.connect(makeOption(option));

        subTopic();
    }

    private void subTopic() {
        if (mqttClient != null && mqttClient.isConnected()) {
            // topic
            Optional.ofNullable(option.getTopics()).ifPresent(ts -> {
                Arrays.stream(ts).forEach(t -> {
                    try {
                        if (t.getHandle() != null) {
                            t.getHandle().newInstance().register(t.getName());
                        }
                        mqttClient.subscribe(t.getName(), 2);
                    } catch (Exception e) {
                        log.info(">>>订阅topic：{} 异常 ", t, e);
                    }
                });
            });
        }
    }

    public void disconnect() {
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                log.info(">>>disconnect 异常", e);
            }
        }
    }

    public void reconnect() {
        if (mqttClient != null && !mqttClient.isConnected()) {
            try {
                mqttClient.reconnect();
            } catch (MqttException e) {
                log.info(">>>reconnect 异常", e);
            }
        }
    }

    public void sendMsg(MqttMsg msg) {
        if (mqttClient == null) {
            throw new RuntimeException("mqtt客户端未启动!");
        }
        // 重连
        if (mqttClient.isConnected()) {
            try {
                mqttClient.reconnect();
            } catch (MqttException e) {
                log.info(">>>reconnect 异常", e);
                throw new RuntimeException(e);
            }

        }
        MqttMessage message = new MqttMessage();
        message.setPayload(msg.getPayload().getBytes(StandardCharsets.UTF_8));
        message.setQos(2);
        try {
            message.setMutable(true);
            mqttClient.publish(msg.getTopic(), message);
        } catch (MqttException e) {
            log.info(">>>sendMsg {}", msg);
        }
    }
}
